﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace Ec.NETCoreFramework.Common.MQ
{
    /// <summary>
    /// 消息消费
    /// </summary>
    public class MsgConsume
    {
        private IConnection _connection = null;
        private IModel _channel = null;
        private string _hostName = string.Empty;
        private string _userName = string.Empty;
        private string _pwd = string.Empty;
        private string _virtualHost = string.Empty;
        public MsgConsume(string hostName, string userName, string pwd, string virtualHost)
        {
            _hostName = hostName;
            _userName = userName;
            _pwd = pwd;
            _virtualHost = virtualHost;
        }
        public TaskInterface taskInterface { get; set; }
        public void Consume(TaskInterface consumeInterface)
        {
            this.taskInterface = consumeInterface;

            var factory = new ConnectionFactory()
            {
                HostName = _hostName,
                Password = _pwd,
                UserName = _userName,
                ClientProvidedName = consumeInterface.QueueName(),
                VirtualHost = _virtualHost
            };
            _connection = factory.CreateConnection();
            _channel = _connection.CreateModel();
            {
                _channel.QueueDeclare(queue: consumeInterface.QueueName(),
                                     durable: true,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);

                _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

                Console.WriteLine(" [*] Waiting for messages.");
                //保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。在这种情况下发送端会尝试把消息发送给下一个not busy的接收端
                var consumer = new EventingBasicConsumer(_channel);
                consumer.Received += (sender, ea) =>
                {
                    var body = ea.Body.ToArray();

                    consumeInterface.ConsumeEvent(body);

                    // Note: it is possible to access the channel via
                    //       ((EventingBasicConsumer)sender).Model here
                    _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                };
                _channel.BasicConsume(queue: consumeInterface.QueueName(),
                                     autoAck: false,
                                     consumer: consumer);

            }


        }
        public bool IsRunning()
        {
            return _channel != null && _channel.IsOpen && _connection != null && _connection.IsOpen;
        }
        /// <summary>
        /// 关闭
        /// </summary>
        public void Close()
        {
            if (_channel != null)
            {
                _channel.Close();
            }
            if (_connection != null)
            {
                _connection.Close();
            }
        }
    }
}
